package com.syf.kit.soa.core.send.protocol.client;

import com.syf.kit.soa.core.SyfRpcMsg;
import com.syf.kit.soa.core.send.protocol.TcpClient;
import com.syf.kit.soa.core.send.protocol.TcpResponseCoordinator;
import com.syf.kit.soa.core.send.protocol.TcpResponseLatch;
import com.syf.kit.soa.core.send.protocol.exception.RpcResponseTimeoutException;
import com.syf.kit.soa.core.send.protocol.handle.RpcDataDecoder;
import com.syf.kit.soa.core.send.protocol.handle.RpcDataEncoder;
import com.syf.kit.soa.core.send.protocol.handle.RpcSendHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.util.concurrent.TimeUnit;

/**
 * 未池化的tcp client，它是短连接，请求完毕就会关闭连接
 */
public class UnPooledTcpClient implements TcpClient {
    private String host;

    private Integer port;

    private Bootstrap bootstrap;

    /**
     * 全局连接超时时间，单位：毫秒
     */
    private long connTimeout;

    public UnPooledTcpClient(String host, int port, long connTimeout) {
        this.host = host;
        this.port = port;
        this.connTimeout = connTimeout;

        initialize();
    }

    private void initialize() {
        this.bootstrap = new Bootstrap();
        EventLoopGroup loopGroup = new NioEventLoopGroup();
        bootstrap.group(loopGroup).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        ChannelPipeline pipeline = channel.pipeline();
                        pipeline.addLast(new RpcDataDecoder());
                        pipeline.addLast(new RpcDataEncoder());
                        pipeline.addLast(new RpcSendHandler());
                    }
                });
    }

    /**
     * 发送数据
     *
     * @param reqMsg     请求数据体
     * @param reqTimeout 服务端响应超时时间
     */
    @Override
    public SyfRpcMsg send(SyfRpcMsg reqMsg, long reqTimeout) {
        ChannelFuture cf = null;
        try {
            cf = bootstrap.connect(host, port);
            if (!cf.await(connTimeout, TimeUnit.MILLISECONDS)) {
                throw new RuntimeException("建立连接超时，目标：[" + host + ":" + port + "]");
            }

            TcpResponseLatch<SyfRpcMsg> sendLatch = TcpResponseCoordinator.getInstance().loadSendLatch(reqMsg.getReqUUID());
            cf.channel().writeAndFlush(reqMsg);
            return sendLatch.getResponse(reqTimeout);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (RpcResponseTimeoutException e) {
            throw new RuntimeException(e);
        } finally {
            if (cf != null) {
                cf.channel().close();
            }
        }
    }

    @Override
    public void close() {

    }
}
